package org.kth.dks.dks_dht;

import org.kth.dks.DKSImpl;
import org.kth.dks.DKSObject;
import org.kth.dks.dks_comm.DKSRef;
import org.kth.dks.dks_node.DKSNode;
import org.kth.dks.util.MathMisc;

class DHTSendStateHandler extends Thread {
    DKSImpl myDKS=null;
    long start;
    long stop;
    DKSRef target;
    boolean cont=true;
    DHTStorage store;

    public DHTSendStateHandler(DKSImpl dks, long strt, long stp, DKSRef trgt,DHTStorage s, boolean leave){
	myDKS=dks;
	start=strt;
	stop=stp;
	store=s;
	target=trgt;
    this.setName(DHTSendStateHandler.class.getName());
    }

    public void run(){
	try {
	    long[] items=store.getAllKeys();

	    for(int i=0;i<items.length;i++){
		if (MathMisc.belongsTo(items[i], start, stop, DKSNode.N)) {
		    DKSObject[] objs=store.lookupItem(items[i]);
		    for(int j=0;j<objs.length;j++)
			store.removeItem(items[i], objs[j]);
		    DHTStateTransferMsg msg=new DHTStateTransferMsg(items[i], objs);
		    myDKS.send(target,msg);
		}
	    }
	} catch(Exception ex) {
	    ex.printStackTrace();
	}

	DHTStateTransferMsg msg=new DHTStateTransferMsg(stop, DHTStateTransferMsg.LASTMSG);
	myDKS.send(target,msg);
    }

    public void finish(){
	cont=false;
    }
}
